package com.csust.common.hive;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.security.UserGroupInformation;

import java.io.IOException;
import java.sql.*;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import static com.csust.common.pojo.SystemConstants.*;
import static com.csust.common.pojo.SystemConstants.PATH_TO_USER2_KEYTAB;

/***
 * 通过kerberos认证的 hive operator
 * @author yhm
 * @date 2021/10/31
 * @descripe hive2.1.1 与 hadoop3.0.0 冲突，所以降低hadoop版本使用hive-jdbc自带的hadoop2（详见maven）。
 * hadoop3可以兼容hadoop2，经测试不影响hdfs operator的使用。
 * 暂定只使用user2操作hive,有拓展需求可以使用其他user
 */
@Slf4j
@Data
public class HiveOp {
    private static final String DRIVER = "org.apache.hive.jdbc.HiveDriver";
    private static final String USERNAME = "";
    private static final String PASSWORD = "";
    private static final String URL = "jdbc:hive2://csust-dataengine-core-1-8a220-0.hde.com:10000/;principal=hive/_HOST@CSUST.COM";

    private Connection conn = null;

    public HiveOp() {
        this(USER2_PRINCIPAL, PATH_TO_USER2_KEYTAB);
    }

    public HiveOp(String userPrincipal, String path2UserKeytab) {
        // 注意，URL中principal中的hostname必须为hiveserver2所在节点的主机名
        // 设置jvm启动时krb5的读取路径参数
        System.setProperty(JVM_KRB5_CONF_PARM, JVM_KRB5_CONF_PARM_VALUE);
        // 配置kerberos认证
        Configuration conf = new Configuration();
        conf.setBoolean("hadoop.security.authorization", true);
        conf.set("hadoop.security.authentication", "kerberos");
        conf.set("hive.security.authentication", "kerberos");
        UserGroupInformation.setConfiguration(conf);
        try {
            UserGroupInformation.loginUserFromKeytab(userPrincipal, path2UserKeytab);
            log.info("Kerberos 身份认证成功");
        } catch (IOException e) {
            log.error("Kerberos 身份认证失败：");
        }
        try {
            Class.forName(DRIVER);
            conn = DriverManager.getConnection(URL, USERNAME, PASSWORD);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    /**
     * 执行hql语句
     * @param hql hql语句
     * @return List<List<String>>
     */
    public List<List<String>> executeHql_LL(String hql) throws SQLException {
        Statement sm = getConn().createStatement();
        ResultSet rs = sm.executeQuery(hql);
        log.info("execute hql:" + hql);
        List<List<String>> list = new ArrayList<List<String>>();
        //获取键名
        ResultSetMetaData md = rs.getMetaData();
        while (rs.next()) {
            ArrayList<String> rowData = new ArrayList<String>();
            for (int i = 1; i <= md.getColumnCount(); i++) {
                rowData.add(rs.getObject(i).toString());
            }
            list.add(rowData);
        }
        sm.close();
        rs.close();
        return list;
    }

    /**
     * 执行hql语句
     * @param hql hql语句
     * @return List<Map<String,String>>
     */
    public List<Map<String,String>> executeHql_LM(String hql) throws SQLException {
        Statement sm = getConn().createStatement();
        log.info("execute hql:" + hql);
        ResultSet rs = sm.executeQuery(hql);
        List<Map<String,String>> list = new ArrayList<Map<String,String>>();
        //获取键名
        ResultSetMetaData md = rs.getMetaData();
        while (rs.next()) {
            Map<String,String> rowData = new HashMap<String,String>();
            for (int i = 1; i <= md.getColumnCount(); i++) {
                rowData.put(md.getColumnName(i), rs.getObject(i).toString());
            }
            list.add(rowData);
        }
        sm.close();
        rs.close();
        return list;
    }

    /**
     * 释放资源
     */
    public void destory() throws SQLException {
        if (conn != null) {
            conn.close();
        }
    }

    public static void main(String[] args) throws Exception {

        HiveOp hiveOp = new HiveOp();
        System.out.println(hiveOp.executeHql_LL("show databases"));
        System.out.println(hiveOp.executeHql_LM("show databases"));
        hiveOp.destory();
    }
}
